package org.geekbang.resilience4j.bulkhead;

import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;

import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;
import io.github.resilience4j.bulkhead.Bulkhead;
import io.github.resilience4j.bulkhead.BulkheadConfig;
import io.github.resilience4j.bulkhead.BulkheadRegistry;
import io.github.resilience4j.bulkhead.ThreadPoolBulkhead;
import io.github.resilience4j.bulkhead.ThreadPoolBulkheadConfig;
import io.github.resilience4j.bulkhead.ThreadPoolBulkheadRegistry;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.BaseFilter;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * resilience4j Bulkhead Semaphore processing.
 * @author leo
 * @date 2021/9/5
 * @since 1.0.0
 */
@Activate(group = PROVIDER,value = {
        BulkheadConstants.PATTERN,
        BulkheadConstants.MAX_CONCURRENT_CALLS,
        BulkheadConstants.MAX_WAIT_MILL_SECONDS,
        BulkheadConstants.MAX_THREAD_POOL_SIZE,
        BulkheadConstants.CORE_THREAD_POOL_SIZE,
        BulkheadConstants.QUEUE_CAPACITY,
        BulkheadConstants.KEEP_ALIVE_DURATION_MILLS})
public class BulkheadFilter implements Filter, BaseFilter.Listener {

    private final Logger logger = LoggerFactory.getLogger(BulkheadFilter.class);

    private final ConcurrentHashMap<String,Bulkhead> semaphore = new ConcurrentHashMap<>(64);

    private final ConcurrentHashMap<String,ThreadPoolBulkhead> threadPool = new ConcurrentHashMap<>(64);



    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        URL url = invoker.getUrl();
        String methodName = invocation.getMethodName();
        BulkheadConstants.Pattern pattern = BulkheadConstants.Pattern.of(url.getMethodParameter(methodName,BulkheadConstants.PATTERN));
        if (pattern == BulkheadConstants.Pattern.SEMAPHORE) {
            return semaphore(invoker, invocation, url, methodName);
        } else if (pattern == BulkheadConstants.Pattern.THREAD_POOL) {
            return fixedThreadPool(invoker, invocation, url, methodName);
        }
        return invoker.invoke(invocation);
    }

    private Result semaphore(Invoker<?> invoker, Invocation invocation, URL url, String methodName) {
       Bulkhead bulkhead = semaphore.computeIfAbsent(methodName,(key) ->{
        //设置最大等待时间(ms)
           long maxWaitMillSeconds = url.getMethodParameter(methodName,BulkheadConstants.MAX_WAIT_MILL_SECONDS,
                   BulkheadConfig.DEFAULT_MAX_WAIT_DURATION.toMillis());
           //设置并发访问量
           int maxConcurrentCalls = url.getMethodParameter(methodName,
                   BulkheadConstants.MAX_CONCURRENT_CALLS,BulkheadConfig.DEFAULT_MAX_CONCURRENT_CALLS);
           BulkheadRegistry registry = BulkheadRegistry.of(BulkheadConfig.custom()
                   .maxWaitDuration(Duration.ofMillis(maxWaitMillSeconds)).maxConcurrentCalls(maxConcurrentCalls)
                   .build());
           Bulkhead bk = registry.bulkhead(BulkheadConstants.SEMAPHORE_FILTER_NAME);
           //监听拒绝访问
           bk.getEventPublisher()
                   .onCallRejected(reject ->
                   onError(new RpcException(RpcException.LIMIT_EXCEEDED_EXCEPTION,"resilience4j bulkhead overhead limited"),
                           invoker, invocation));
           return bk;
        });

        return bulkhead.executeSupplier(()-> invoker.invoke(invocation));
    }

    private Result fixedThreadPool(Invoker<?> invoker, Invocation invocation, URL url, String methodName) {
       ThreadPoolBulkhead bulkhead = threadPool.computeIfAbsent(methodName,(key) ->{
            //设置最大线程数
            int maxThreadPoolSize = url.getMethodParameter(methodName,BulkheadConstants.MAX_THREAD_POOL_SIZE,
                    Runtime.getRuntime().availableProcessors());
            //核心最大线程数
            int coreThreadPoolSize = url.getMethodParameter(methodName,BulkheadConstants.CORE_THREAD_POOL_SIZE,
                    Runtime.getRuntime().availableProcessors() - 1);
            //队列大小
            int queueCapacity = url.getMethodParameter(methodName,BulkheadConstants.QUEUE_CAPACITY,
                    100);
            //keep alive
            int keepAliveDurationMills = url.getMethodParameter(methodName,BulkheadConstants.KEEP_ALIVE_DURATION_MILLS,
                    20);
            ThreadPoolBulkheadConfig config = ThreadPoolBulkheadConfig.custom()
                    .maxThreadPoolSize(maxThreadPoolSize)
                    .coreThreadPoolSize(coreThreadPoolSize)
                    .queueCapacity(queueCapacity)
                    .keepAliveDuration(Duration.ofMillis(keepAliveDurationMills))
                    .build();

            // Create a BulkheadRegistry with a custom global configuration
            ThreadPoolBulkheadRegistry registry = ThreadPoolBulkheadRegistry.of(config);

            // Get or create a ThreadPoolBulkhead from the registry -
            // bulkhead will be backed by the default config
            ThreadPoolBulkhead bk = registry.bulkhead(BulkheadConstants.THREAD_POOL_FILTER_NAME);

            //监听拒绝访问
           bk.getEventPublisher().onCallRejected(reject ->
                    onError(new RpcException(RpcException.LIMIT_EXCEEDED_EXCEPTION,"resilience4j bulkhead overhead limited"),
                            invoker, invocation));
            return bk;
        });
        return bulkhead.executeSupplier(()-> invoker.invoke(invocation)).toCompletableFuture().join();
    }

    @Override
    public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
        if (logger.isDebugEnabled()){
            logger.debug("response :{}",appResponse);
        }
    }

    @Override
    public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
        if (t instanceof RpcException){
            RpcException rpcException = (RpcException) t;
            //超限
            if (rpcException.isLimitExceed()){
                logger.warn("url:{} limit exceed!",invoker.getUrl().toServiceString());
                throw rpcException;
            }
        }
        logger.error("url:{}, invoker error:{}",invoker.getUrl(),t);
        throw new RpcException(t);
    }
}
